package com.jsj.subway.ai.demo.service.impl;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.itextpdf.html2pdf.ConverterProperties;
import com.itextpdf.html2pdf.HtmlConverter;
import com.itextpdf.kernel.pdf.PdfDocument;
import com.itextpdf.kernel.pdf.PdfWriter;
import com.itextpdf.layout.font.FontProvider;
import com.jsj.subway.ai.agent.chat.IntentRecognitionService;
import com.jsj.subway.ai.agent.rag.KnowledgeBaseFilesService;
import com.jsj.subway.ai.common.ApiException;
import com.jsj.subway.ai.common.R;
import com.jsj.subway.ai.common.Role;
import com.jsj.subway.ai.demo.constants.Constants;
import com.jsj.subway.ai.demo.dto.AlarmMessage;
import com.jsj.subway.ai.demo.dto.ChatDTO;
import com.jsj.subway.ai.demo.service.DemoService;
import com.jsj.subway.ai.largemodel.AliClient;
import com.jsj.subway.ai.largemodel.config.AliConfig;
import com.jsj.subway.ai.largemodel.models.AliModelCompletionResponse;
import com.jsj.subway.ai.largemodel.models.Message;
import com.jsj.subway.ai.largemodel.models.MessageEvents;
import com.jsj.subway.ai.utils.LocalCache;
import com.jsj.subway.ai.utils.NotifyServer;
import com.jsj.subway.ai.utils.SSEContextHolder;
import com.vladsch.flexmark.html.HtmlRenderer;
import com.vladsch.flexmark.util.data.MutableDataSet;
import com.vladsch.flexmark.parser.Parser;
import lombok.extern.slf4j.Slf4j;
import cn.hutool.core.lang.Pair;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import reactor.core.publisher.Sinks;

import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

@Slf4j
@Service
public class DemoServiceImpl implements DemoService {
    private final Object lock = new Object();
    @Value("${kb.systemPrompt2}")
    private String systemPrompt;

    @Value("${pdf.path}")
    private String pdfPath;

    @Value("${pdf.base_uri}")
    private String baseUri;

    @Autowired
    private AliClient aliClient;

    @Autowired
    private IntentRecognitionService intentRecognition;

    @Autowired
    private LocalCache localCache;

    @Autowired
    NotifyServer notifyServer;

    @Autowired
    private AliConfig aliConfig;

    @Autowired
    private KnowledgeBaseFilesService knowledgeBaseFilesService;


    public Deque<ChatDTO.MessageRecord> getMessageRecord(Long key) {
        Deque<ChatDTO.MessageRecord> messageRecordsFromCache = localCache.getMessageRecordsFromCache(key);
        return messageRecordsFromCache;
    }

    @Override
    public String begin(Integer key) {
        localCache.removeRecord(1L);
        localCache.setCounter(key);
        String sseId = UUID.randomUUID().toString();
        SseEmitter sseEmitter = localCache.getEmitter(sseId);
        AlarmMessage alarmMessage = Constants.map.get(key);
        String date = DateUtil.format(new Date(), "yyyy-MM-dd HH:mm:ss");
        String station = Constants.list.stream()
                .skip(new Random().nextInt(Constants.list.size()))
                .findFirst()
                .orElse("王家港");
        AlarmMessage build = AlarmMessage.builder()
                .alarmType(alarmMessage.getAlarmType())
                .line(alarmMessage.getLine())
                .type(alarmMessage.getType())
                .name(alarmMessage.getName())
                .describe(alarmMessage.getDescribe())
                .station(station)
                .createTime(date)
                .SSEId(sseId)
                .firstMessage(StrUtil.format(alarmMessage.getFirstMessage(), station, date))
                .build();
        log.info(build.toString());
        localCache.setSharedMap(build);
        notifyServer.onMessage(JSONUtil.toJsonStr(build));

        //调用大模型
//        ChatDTO.MessageRequest request = new ChatDTO.MessageRequest();
//        request.setContent("你好，可以给我写一手诗么");
//        ChatDTO.MessageRecord messageRecord = this.sessionChat(request, sseId);
        return null;
    }

    @Override
    public ChatDTO.MessageRecord sessionChatByStream(ChatDTO.MessageRequest messageRequest) {

        AtomicBoolean pdfFlag = new AtomicBoolean(false);
        Long sessionId = messageRequest.getSessionId();
        String sseId = SSEContextHolder.get();
        if (Objects.isNull(messageRequest.getContent())) {
            throw new ApiException("内容不能为空！");
        }

        LinkedList<Message> sessionRecords = localCache.getSessionMessagedForApi(sessionId);
        Message userMessage = new Message();
        int size = sessionRecords.size();

        List<Message> messageContext = new LinkedList<>();


        //表明首次进入
        if (1 == messageRequest.getIsFirst()) {
            userMessage = Message.toMessage(messageRequest.getContent(), "user");
            messageContext.add(userMessage);
        } else {
            userMessage = Message.toMessage(messageRequest.getContent(), Role.USER.getLabel());
            messageContext = new LinkedList<>(sessionRecords.subList(size > aliConfig.getContextRound() ? size - aliConfig.getContextRound() : 0, size));

            //意图识别
            Pair<Integer, List<Message>> integerListPair = intentRecognition.intentRecognition(Collections.singletonList(userMessage),sessionId );
            //Pair<Integer, List<Message>> integerListPair  = new Pair<>(2,Collections.singletonList(userMessage));
            if (Objects.isNull(integerListPair)) {
                //意图报错
                log.error("Error in sessionChat: ：{}", "意图识别报错");
                ChatDTO.MessageRecord messageRecord = new ChatDTO.MessageRecord();
                messageRecord.setContent("意图识别报错");
                messageRecord.setRole(Role.ASSISTANT.getLabel());
                messagePushAndContinue(sseId, R.ok(messageRecord));
                completeSse(sseId);
                return null;
            }
            if (0 == integerListPair.getKey()) {
                List<Message> value = integerListPair.getValue();
                if (CollectionUtils.isEmpty(value)) {
                    ChatDTO.MessageRecord messageRecord = new ChatDTO.MessageRecord();
                    messageRecord.setContent("意图识别报错11");
                    messageRecord.setRole(Role.ASSISTANT.getLabel());
                    log.error("Error in sessionChat: ：{}", "意图识别报错11");
                    messagePushAndContinue(sseId, R.ok(messageRecord));
                    completeSse(sseId);
                    return null;
                } else {
                    Message message = value.get(0);
                    ChatDTO.MessageRecord messageRecord = new ChatDTO.MessageRecord();
                    messageRecord.setContent(message.getContent());
                    messageRecord.setRole(Role.ASSISTANT.getLabel());

                    log.error("Error in sessionChat: ：{}", message.getContent());
                    messagePushAndContinue(sseId, R.ok(messageRecord));
                    completeSse(sseId);
                    return null;
                }
            }
            if (1 == integerListPair.getKey()) {
                List<Message> value = integerListPair.getValue();
                ChatDTO.MessageRecord result = new ChatDTO.MessageRecord();
                result.setContent("正在执行重合闸操作，等待执行结果 \n " + value.get(0).getContent() );
                result.setRole(Role.ASSISTANT.getLabel());
                messagePushAndContinue(sseId, R.ok(result));
                //组装模拟用户信息
                messageContext.add(Message.toMessage(systemPrompt, "system"));
                messageContext.addAll(value);

                int counter = localCache.getCounter();
                String s = Constants.resultMap.get(counter);
                List<String> list = Arrays.asList(s.split(","));

                //操作重合闸 回调信息
                AlarmMessage sharedMap = localCache.getSharedMap();
                sharedMap.setCreateTime(DateUtil.format(new Date(),"yyyy-MM-dd HH:mm:ss"));
                sharedMap.setAlarmType("信号");
                sharedMap.setDescribe(list.get(0));
                sharedMap.setSSEId("");
                sharedMap.setFirstMessage("");
                notifyServer.onMessage(JSONUtil.toJsonStr(sharedMap));


                //临闸
                sharedMap = localCache.getSharedMap();
                sharedMap.setCreateTime(DateUtil.format(new Date(),"yyyy-MM-dd HH:mm:ss"));
                sharedMap.setAlarmType("信号");
                sharedMap.setDescribe(list.get(1));
                sharedMap.setSSEId("");
                sharedMap.setFirstMessage("");
                sharedMap.setStation(changeStation(sharedMap.getStation()));

                notifyServer.onMessage(JSONUtil.toJsonStr(sharedMap));
                //completeSse(sseId);
                //return null;
            }
            //不相关
            if (2 == integerListPair.getKey()) {

                Pair<Integer, List<Message>> integerListPairPDF = intentRecognition.intentRecognitionCreatePDF(Collections.singletonList(userMessage), sessionId);
                if (1 == integerListPairPDF.getKey()) {

                    messageContext.addAll(integerListPairPDF.getValue());
                    pdfFlag.set(true);

                } else { //除了1 其余都不生成文档。
                    List<Message> value = integerListPair.getValue();
                    //需要携带 prompt
                    messageContext.add(Message.toMessage(systemPrompt, "system"));
                    messageContext.addAll(value);
                }


            }
        }

        //RAG
        //messageContext = knowledgeBaseFilesService.queryKbBase(messageContext, false);

        log.info("sessionChat 当前信息：{}", messageContext.toString());
        Sinks.Many<ChatDTO.MessageRecord> sink = Sinks.many().multicast().onBackpressureBuffer();

        AtomicBoolean flag = new AtomicBoolean(true);
        AtomicBoolean hasError = new AtomicBoolean(false);
        List<Message> allMessages = new ArrayList<>();
        ChatDTO.MessageRecord ur = ChatDTO.toMessageRecord(Collections.singletonList(userMessage), false);
        aliClient.completionsChatByStream(messageContext).subscribe(chatResult -> {
            if (chatResult.getErr_code() == 0) {

                ChatDTO.MessageRecord result = ChatDTO.toMessageRecord(chatResult.toMessageByStream(), true);
                allMessages.addAll(chatResult.toMessageByStream());
                //localCache.addSessionMessage(sessionId, result);
                //saveChatRecord(sessionId, ur, result);
                messagePushAndContinue(sseId, R.ok(result));
            } else {
                log.info("会话失败 sseId={},message={}", sseId, chatResult.getErr_msg());
                messagePushAndContinue(sseId, R.failed(chatResult.getErr_msg()));
            }
        }, error -> {
            hasError.set(true);
            log.error("Error in sessionChat: ", error);
            messagePushAndContinue(sseId, R.failed(error.getMessage()));
            //sink.tryEmitNext(new ChatDTO.MessageRecord());
        }, () -> {
            if (flag.get()) {
                localCache.addMessageRecordToCache(sessionId, ur);
                flag.set(false);
            }
            if (hasError.get()) {
                completeSse(sseId);
                return;
            }
            if (pdfFlag.get()) {
                ChatDTO.MessageRecord messageRecord = new ChatDTO.MessageRecord();
                messageRecord.setRole(Role.ASSISTANT.getLabel());
                StringBuffer content = new StringBuffer();
                allMessages.stream().map(Message::getContent).forEach(content::append);
                String urlByContent = getUrlByContent(content.toString());
                messageRecord.setContent("下载地址为：<a target=\"blank\" href='"+urlByContent+"'>下载</a>");
                messagePushAndContinue(sseId, R.ok(messageRecord));
            }
            //保存后 输出需要存缓存的record
            ChatDTO.MessageRecord messageRecord = saveChatRecordByStream(sessionId, allMessages, messageRequest);
            localCache.addMessageRecordToCache(sessionId, messageRecord);
            completeSse(sseId);
            sink.tryEmitComplete();
        });
        return null;
    }


    @Override
    public ChatDTO.MessageRecord sessionChat(ChatDTO.MessageRequest messageRequest) {
        if (Objects.isNull(messageRequest.getContent())) {
            throw new ApiException("内容不能为空！");
        }

        Message userMessage = Message.toMessage(messageRequest.getContent(), Role.USER.getLabel());
        List<Message> messageContext = new LinkedList<>();
        messageContext.add(userMessage);


        log.info("sessionChat 当前信息：{}", messageContext.toString());

        try {
            AliModelCompletionResponse chatResult = aliClient.completionsChat(messageContext);
            if (chatResult.getErr_code() == 0) {
                List<Message> allMessages = chatResult.toMessage(); // Assuming chatResult.toMessageByStream() converts to List<Message>
                ChatDTO.MessageRecord result = ChatDTO.toMessageRecord(allMessages, false);
                return result;
            } else {
                log.info("会话失败 ,message={}", chatResult.getErr_msg());
                return null;
            }
        } catch (Exception e) {
            log.error("Error in sessionChat: ", e);
            return null;
        }
    }


    @Override
    public SseEmitter connect(HttpServletRequest request, HttpServletResponse response) {
        String sseId = request.getParameter("SSE");
        //String sseId = SSEContextHolder.get();
        SseEmitter sseEmitter = localCache.getEmitter(sseId);
        try {
            sseEmitter.send(SseEmitter.event()
                    .id(MessageEvents.ID_SIGNAL)
                    .data(sseId, MediaType.TEXT_EVENT_STREAM)
                    .reconnectTime(1000));
            log.info("获取SSE连接成功！");
            return sseEmitter;
        } catch (IOException e) {
            e.printStackTrace();
            log.error("获取SSE连接失败！");
            return null;
        }
    }

    @Override
    public SseEmitter connect(HttpServletResponse response) {
        String sseId = UUID.randomUUID().toString();
        SseEmitter sseEmitter = localCache.getEmitter(sseId);
        response.setContentType("text/event-stream;charset=utf-8");
        try {
            sseEmitter.send(SseEmitter.event()
                    .id(MessageEvents.ID_SIGNAL)
                    .data(sseId, MediaType.TEXT_EVENT_STREAM)
                    .reconnectTime(1000));
            log.info("获取SSE连接成功！");
            return sseEmitter;
        } catch (IOException e) {
            e.printStackTrace();
            log.error("获取SSE连接失败！");
            return null;
        }
    }

    private void completeSse(String sseId) {
        synchronized (lock) {
            if (sseId == null) {
                log.error("sseId 为空 无法关闭连接");
                return;
            }

            SseEmitter sseEmitter = localCache.getEmitter(sseId);
            if (sseEmitter == null) {
                log.error("找不到 sseEmitter，无法关闭连接，sseId={}", sseId);
                return;
            }

            try {
                sseEmitter.send(SseEmitter.event()
                        .id(MessageEvents.DONE_SIGNAL)
                        .data(MessageEvents.DONE_SIGNAL, MediaType.TEXT_EVENT_STREAM)
                        .reconnectTime(3000));
                sseEmitter.complete();
                log.info("关闭 sseId={}", sseId);
                localCache.invalidateEmitter(sseId);
            } catch (Exception e) {
                log.error("关闭连接失败 sseId={}, 错误信息：{}", sseId, e.getMessage());
            }
        }
    }


    private boolean messagePushAndContinue(String sseId, Object content) {
        synchronized (lock) {
            if (sseId == null) {
                log.error("sseId 为空 无法推送消息");
                return false;
            }

            SseEmitter sseEmitter = localCache.getEmitter(sseId);
            if (sseEmitter == null) {
                log.error("找不到 sseEmitter，无法推送消息，sseId={}", sseId);
                return false;
            }

            try {
                sseEmitter.send(SseEmitter.event()
                        .id(MessageEvents.CONTENT_SIGNAL)
                        .data(JSONUtil.toJsonStr(content), MediaType.TEXT_EVENT_STREAM)
                        .reconnectTime(3000));
                log.info("发送消息到 sseId={}", sseId);
            } catch (IOException e) {
                log.error("发送消息失败 sseId={}, 错误信息：{}", sseId, e.getMessage());
                return false;
            }

            return true;
        }

    }

    private boolean messagePushAndClose(String sseId, Object content) {
        synchronized (lock) {
            if (sseId == null) {
                log.error("sseId 为空 无法推送消息");
                return false;
            }
            SseEmitter sseEmitter = localCache.getEmitter(sseId);
            try {
                sseEmitter.send(SseEmitter.event()
                        .id(MessageEvents.CONTENT_SIGNAL)
                        .data(JSONUtil.toJsonStr(content), MediaType.TEXT_EVENT_STREAM)
                        .reconnectTime(1000));
                sseEmitter.complete();
                log.info("close sseId={}", sseId);
                localCache.invalidateEmitter(sseId);
            } catch (IOException e) {
                e.printStackTrace();
                return false;
            }
            return true;
        }
    }

    public ChatDTO.MessageRecord saveChatRecordByStream(Long sessionId, List<Message> allMessages, ChatDTO.MessageRequest messageRequest) {
        ChatDTO.MessageRecord returnMessage = new ChatDTO.MessageRecord();
        //置换成没有增强的问话
        //ur.setContent(messageRequest.getContent());
//        if (1 == messageRequest.getHasFile()) {
//            KnowledgeBaseFileDTO.QueryDTO fileBySessionId = knowledgeBaseFilesService.getFileBySessionId(messageRequest.getSessionId());
//            StringBuffer stringBuffer = new StringBuffer();
//            stringBuffer.append(fileBySessionId.getFileName()).append(" \\n\\n ").append(ur.getContent());
//            ur.setContent(stringBuffer.toString());
//        }

        if (CollectionUtil.isEmpty(allMessages)) {
            return new ChatDTO.MessageRecord();
        }
        StringBuffer content = new StringBuffer();
        allMessages.stream().map(Message::getContent).forEach(content::append);
        Message message = Message.toMessage(content.toString(), Role.ASSISTANT.getLabel());
        returnMessage.setMessages(Collections.singletonList(message));
        returnMessage.setRole(Role.ASSISTANT.getLabel());
        returnMessage.setType(1);
        returnMessage.setContent(content);
        returnMessage.setTime(DateUtil.format(new Date(), "yyyy-MM-dd HH:mm:ss"));
        return returnMessage;
    }


    private byte[] convertMarkdownToPdf(String markdownContent) throws Exception {
        // 将Markdown转换为HTML
        MutableDataSet options = new MutableDataSet();
        Parser parser = Parser.builder(options).build();
        HtmlRenderer renderer = HtmlRenderer.builder(options).build();
        String htmlContent = renderer.render(parser.parse(markdownContent));
        log.info("11");
        // 使用iText将HTML转换为PDF
        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
        PdfWriter writer = new PdfWriter(outputStream);
        PdfDocument pdf = new PdfDocument(writer);
//        Document document = new Document(pdf);
        ConverterProperties converterProperties = new ConverterProperties();

        Pattern pattern = Pattern.compile("<p>.*?【.*?】.*?</p>");
        Matcher matcher = pattern.matcher(htmlContent);
        if (matcher.find()) {
            String targetLine = matcher.group();
            String centeredLine = targetLine.replace("<p>", "<p style=\"text-align:center;\">");
            htmlContent = htmlContent.replace(targetLine, centeredLine);
        }
//        String title = "故障分析";
//        Paragraph titleParagraph = new Paragraph(title)
//                .setTextAlignment(TextAlignment.CENTER)
//                .setFontSize(14); // 设置字体大小
//        document.add(titleParagraph);

        log.info("111");
        log.info(getClass().getClassLoader().getResource("fonts/NotoSansCJKsc-Regular.otf").toString());
        Path fontPath = Paths.get("/opt/fonts/NotoSansCJKsc-Regular.otf");

        // 设置字体file路径
        //Path fontPath = Paths.get(Objects.requireNonNull(getClass().getClassLoader().getResource("fonts/NotoSansCJKsc-Regular.otf")).toURI());
        log.info(fontPath.toString());
        FontProvider fontProvider = new FontProvider();
        fontProvider.addFont(fontPath.toString());
        converterProperties.setFontProvider(fontProvider);

        HtmlConverter.convertToPdf(htmlContent, pdf, converterProperties);

        return outputStream.toByteArray();

    }

    public void savePdfFile(byte[] pdfBytes, String outputPath) throws IOException {
        Path path = Paths.get(outputPath);
        Files.createDirectories(path.getParent()); // 创建目录
        Files.write(path, pdfBytes);
    }

    @Override
    public String getUrlByContent(String content) {
        try {
            byte[] bytes = this.convertMarkdownToPdf(content);

            log.info("1");
            // 获取当前时间并格式化
            LocalDateTime now = LocalDateTime.now();
            DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMdd_HHmmss");
            String formattedDateTime = now.format(formatter);
            String fileName = "故障报告_" + formattedDateTime + ".pdf";
            String outputPath = Paths.get(pdfPath, fileName).toString();
            this.savePdfFile(bytes, outputPath);

            return baseUri + "/pdf/" + fileName;

        } catch (Exception e) {
            log.error(e.getMessage());
            log.info(e.getMessage());
        }
        return null;
    }

    private String changeStation(String station){

        return  Constants.list.stream()
                .filter(s -> !s.equals(station))
                .skip(new Random().nextInt(Constants.list.size()))
                .findFirst()
                .orElse("王家港");
    }
}
